

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ServerLoad;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Row;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import com.ls.fw.data.hbase.bean.HbaseStatus;
import com.ls.fw.data.hbase.bean.Record;
import com.ls.fw.data.hbase.client.HBaseClient;
import com.ls.fw.data.hbase.helper.HbaseHelper;
import com.ls.fw.data.hbase.support.HBaseTable;
import com.ls.fw.data.hbase.support.TableName;

public class Main {
	
	HBaseClient client;
	String name = "test";
//	public static void main(String[] args) {
//		HBaseClient client = HBaseClient.Builder.create().set(HBaseClient.Parameter.ZK_QUORUM, QuorumHost)
//				.setLong(HBaseClient.Parameter.ZK_PROPERTY_CLIENT_PORT, QuorumPort).build();
//		TableName tableName = TableName.valueOf("");
//		client.admin().createTable(tableName );
//		client.admin().addColumnFamily(tableName, "F", Compression.Algorithm.NONE);
//		client.admin().updateCompression(tableName, "F", Compression.Algorithm.LZ4);
//		HBaseTable table = client.table(tableName);
//		table.put(new Put("ROW".getBytes()).addColumn("F".getBytes(), "FQ".getBytes(), "VAL".getBytes()));
//		table.get(new Get("ROW".getBytes()));
//		List<Row> asyncPuts = new ArrayList<>();
//		List<Row> fastPuts = new ArrayList<>();
//		for (int i = 0; i < 10; i++) {
//			asyncPuts.add(new Put(("ASYNC" + i).getBytes()).addColumn("F".getBytes(), "FQ".getBytes(), "V".getBytes()));
//			fastPuts.add(new Put(("FAST" + i).getBytes()).addColumn("F".getBytes(), "FQ".getBytes(), "V".getBytes()));
//		}
//		List<TableName> tableNames = new ArrayList<TableName>();
//		tableNames.parallelStream().forEach(tableName -> {
//			HBaseTable table = client.table(tableName);
//			client.fast(tableName).go(fastPuts);
//			client.async(tableName).batch(asyncPuts, result -> System.out.println("OK"));
//		});
//
//		client.table(HTableModel.tableName(Account.class)).model().scan(new Scan("STARTKEY".getBytes()))
//				.<Account> forEach(account -> System.out.println(account.username()));
//
//		client.admin().deleteColumnFamily(tableName, "F");
//		client.admin().truncateTable(tableName);
//		client.admin().deleteTable(tableName);
//	}
	
	@Before
	public void setup(){
		String QuorumHost = "node1,node3,node4";
		Integer QuorumPort = 2181;
		client = HBaseClient.Builder.create().set(HBaseClient.Parameter.ZK_QUORUM, QuorumHost )
				.setInt(HBaseClient.Parameter.ZK_PROPERTY_CLIENT_PORT, QuorumPort ).build();
	}
	
	@After
	public void close(){
		TableName tableName = TableName.valueOf(name);
		if(client.admin().tableExists(tableName)){
			client.admin().deleteColumnFamily(tableName, "F");
			client.admin().truncateTable(tableName);
			client.admin().deleteTable(tableName);
		}
		client.close();
	}
	
	@Test
	public void create(){
		TableName tableName = TableName.valueOf(name);
		client.admin().createTable(tableName );
		client.admin().addColumnFamily(tableName, "F", Compression.Algorithm.NONE);
		client.admin().updateCompression(tableName, "F", Compression.Algorithm.LZ4);
	
	}
	

	public List<HbaseStatus> getMasterStatus() throws MasterNotRunningException, ZooKeeperConnectionException, IOException {
		List<HbaseStatus> list = new ArrayList<HbaseStatus>();
		//System.out.println("getMasterStatus hbase.master："+hbaseTableManager.getConfiguration().get("hbase.master"));
		HBaseAdmin hBaseAdmin = null;
		hBaseAdmin = (HBaseAdmin) client.admin().admin();
		//master
		//System.out.println("isAborted:"+hBaseAdmin.isAborted());
		//System.out.println("isMasterRunning:"+hBaseAdmin.isMasterRunning());
		ServerName ma = hBaseAdmin.getClusterStatus().getMaster();
		ServerLoad serverLoad = hBaseAdmin.getClusterStatus().getLoad(ma);
		//hBaseAdmin.isMasterRunning()
		if (serverLoad != null) {
			 long startcode = ma.getStartcode();
				 list.add(new HbaseStatus(ma.getHostname(), HostUtil
						.getIPByHostname(ma.getHostname())
						+ ":"
						+ ma.getPort(), "master", serverLoad.getUsedHeapMB(),
						serverLoad.getMaxHeapMB(), serverLoad.getMemstoreSizeInMB(), !hBaseAdmin.isMasterRunning(),
						DateTimeUtil.getFormatDate(startcode))); 
		} else {
			 long startcode = ma.getStartcode();
			 list.add(new HbaseStatus(ma.getHostname(), HostUtil.getIPByHostname(ma
						.getHostname()) + ":" + ma.getPort(), "master",DateTimeUtil.getFormatDate(startcode), !hBaseAdmin.isMasterRunning()));
		}
		Collection<ServerName> v1 = hBaseAdmin.getClusterStatus().getBackupMasters();
		if(v1!=null){
			for (ServerName serverName : v1) {
				ServerLoad sl = hBaseAdmin.getClusterStatus().getLoad(serverName);
				if (sl != null) {
					long startcode = serverName.getStartcode();
					list.add(new HbaseStatus(serverName.getHostname(), HostUtil
							.getIPByHostname(serverName.getHostname())
							+ ":"
							+ serverName.getPort(), "backupMaster", sl.getUsedHeapMB(),
							sl.getMaxHeapMB(), sl.getMemstoreSizeInMB(), false,
							DateTimeUtil.getFormatDate(startcode)));
				}else {
					long startcode = serverName.getStartcode();
					list.add(new HbaseStatus(serverName
							.getHostname(), HostUtil.getIPByHostname(serverName
							.getHostname()) + ":" + serverName.getPort(), "backupMaster",DateTimeUtil.getFormatDate(startcode),true));
				}
			}
		}
		try {
			if (hBaseAdmin != null) {
				hBaseAdmin.close();
			}
			hBaseAdmin = null;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return list;
	}


	public List<HbaseStatus> getRegionServerStatus() throws IOException {
		List<HbaseStatus> list = new ArrayList<HbaseStatus>();
		HBaseAdmin hBaseAdmin =  (HBaseAdmin) client.admin().admin();
		//当前活着的节点
		Collection<ServerName> v1 = hBaseAdmin.getClusterStatus().getServers();
		for (ServerName serverName : v1) {
			ServerLoad sl = hBaseAdmin.getClusterStatus().getLoad(serverName);
			if (sl != null) {
				long startcode = serverName.getStartcode();
				list.add(new HbaseStatus(serverName.getHostname(), HostUtil
						.getIPByHostname(serverName.getHostname())
						+ ":"
						+ serverName.getPort(), "regionServer", sl.getUsedHeapMB(),
						sl.getMaxHeapMB(), sl.getMemstoreSizeInMB(), false,
						DateTimeUtil.getFormatDate(startcode)));
			}else {
				long startcode = serverName.getStartcode();
				list.add(new HbaseStatus(serverName
						.getHostname(), HostUtil.getIPByHostname(serverName
						.getHostname()) + ":" + serverName.getPort(), "regionServer",DateTimeUtil.getFormatDate(startcode),true));
			}
		}
		
		//当前已经停止的节点
		Collection<ServerName> v = hBaseAdmin.getClusterStatus()
				.getDeadServerNames();
		for (ServerName serverName : v) {
			ServerLoad sl = hBaseAdmin.getClusterStatus().getLoad(serverName);
			if (sl != null) {
				long startcode = serverName.getStartcode();
				list.add(new HbaseStatus(serverName.getHostname(), HostUtil
						.getIPByHostname(serverName.getHostname())
						+ ":"
						+ serverName.getPort(), "regionServer", sl.getUsedHeapMB(),
						sl.getMaxHeapMB(), sl.getMemstoreSizeInMB(), true,
						DateTimeUtil.getFormatDate(startcode)));
			} else {
				long startcode = serverName.getStartcode();
				list.add(new HbaseStatus(serverName
						.getHostname(), HostUtil.getIPByHostname(serverName
						.getHostname()) + ":" + serverName.getPort(), "regionServer",DateTimeUtil.getFormatDate(startcode),true));
			}
		}
		try {
			if (hBaseAdmin != null) {
				hBaseAdmin.close();
			}
			hBaseAdmin = null;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return list;
	}
	@Test
	public void add(){
		TableName tableName = TableName.valueOf(name);
		HBaseTable table = client.table(tableName);
		table.put(new Put("ROW".getBytes()).addColumn("F".getBytes(), "FQ".getBytes(), "VAL".getBytes()));
		table.get(new Get("ROW".getBytes()));
		List<Row> asyncPuts = new ArrayList<>();
		List<Row> fastPuts = new ArrayList<>();
		for (int i = 0; i < 10; i++) {
			asyncPuts.add(new Put(("ASYNC" + i).getBytes()).addColumn("F".getBytes(), "FQ".getBytes(), "V".getBytes()));
			fastPuts.add(new Put(("FAST" + i).getBytes()).addColumn("F".getBytes(), "FQ".getBytes(), "V".getBytes()));
		}

		client.fast(tableName).go(fastPuts);
		client.async(tableName).batch(asyncPuts, result -> System.out.println("OK"));

	}
	/**
	 * @throws Exception
	 */
	@Test
	public void test1() throws Exception{
		TableName tableName = TableName.valueOf("base_db_1474699732127_hbase_tst_1479017070811");
		System.out.println(client.table(tableName));
		Scan scan = new Scan();
		scan.setMaxResultSize(10);
//		scan.setStartRow(Bytes.toBytes("00000d5_9223370557826740799_265306"));
//		scan.setStopRow(Bytes.toBytes("7fffa31_9223370557826739067_266303"));
		scan.setStartRow(Bytes.toBytes("7fffa31_9223370557826739067_266303"));
		scan.setStopRow(Bytes.toBytes("00000d5_9223370557826740799_265306"));
		ResultScanner r = client.table(tableName).scan(scan );
		for (Result result : r) {
			Record r1 = HbaseHelper.parse(result, null);
			System.out.println(r1.getId());
		}
	}
	@Test
	public void test() throws Exception{
		TableName tableName = TableName.valueOf("base_db_1474699732127_hbase_tst_1479017070811");
		System.out.println(client.table(tableName));
		Scan scan = new Scan();
		scan.setMaxResultSize(1);
		ResultScanner r = client.table(tableName).scan(scan );
		Record r1 = HbaseHelper.parse(r.next(), null);
		System.out.println(r1.getId());
		scan.setMaxResultSize(1);
		scan.setReversed(false);
//		00000d5_9223370557826740799_265306
		ResultScanner r12 = client.table(tableName).scan(scan );
		Record r11 = HbaseHelper.parse(r12.next(), null);
		System.out.println(r11.getId());
		scan.setMaxResultSize(1);
		scan.setReversed(true);
//		7fffa31_9223370557826739067_266303
		  r12 = client.table(tableName).scan(scan );
		  r11 = HbaseHelper.parse(r12.next(), null);
		System.out.println(r11.getId());
		scan.setMaxResultSize(1);
//		7fffa31_9223370557826739067_266303
		  r12 = client.table(tableName).scan(scan );
		  r11 = HbaseHelper.parse(r12.next(), null);
		System.out.println(r11.getId());
	}
}
